﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading.Tasks;
using ElasticSearchRealize.Model;
using Nest;
using Newtonsoft.Json;

namespace ElasticSearchRealize.QueryProvider
{
    /// <summary>
    /// 索引操作提供者基类
    /// </summary>
    /// <typeparam name="TEntity"></typeparam>
    public abstract class IndexProvider<TEntity> : IIndexProvider<TEntity>
        where TEntity : class, IDocumentIndexIdentity
    { 
        /// <summary>
        /// The create index lock
        /// </summary>
        private readonly static object _createIndexLock = new object();

        /// <summary>
        /// 获取索引的类型属性映射描述
        /// </summary>
        /// <returns></returns>
        protected abstract Func<PropertiesDescriptor<TEntity>, IPromise<IProperties>> GetIndexPropertiesDescriptor();

        /// <summary>
        /// 获取索引的设置描述
        /// </summary>
        /// <returns></returns>
        protected virtual Func<IndexSettingsDescriptor, IPromise<IIndexSettings>> GetIndexSettingsDecriptor()
        {
            return s => s.NumberOfShards(5).NumberOfReplicas(1);
        }

        /// <summary>
        /// Creates the index.
        /// </summary>
        /// <param name="indexName">Name of the index.</param>
        public virtual void CreateIndex(string indexName)
        {
            if (!ElasticSearchHelper.IfIndexExists(indexName))
            {
                lock (_createIndexLock)
                {
                    if (!ElasticSearchHelper.IfIndexExists(indexName))
                    {
                        var client = ElasticSearchHelper.GetElasticClient();
                        var resp = client.Indices.Create(indexName, d => d.Settings(GetIndexSettingsDecriptor())
                                        .Map<TEntity>(md => md.Properties(GetIndexPropertiesDescriptor())));
                        if (resp.IsValid)
                        {
                            ElasticSearchHelper.AddIndexName(indexName);
                            return;
                        }
                            
                        throw new Exception($"创建索引失败，IndexName：{indexName}，Error：{resp.GetErrorString()}", resp.OriginalException);
                    }
                }
            }
        }

        /// <summary>
        /// 更新索引的类型字段映射
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <exception cref="Exception">更新索引映射失败</exception>
        public virtual void UpdateIndexMapping(string indexName)
        {
            if (ElasticSearchHelper.IfIndexExists(indexName))
            {
                var client = ElasticSearchHelper.GetElasticClient();
                var resp = client.Map<TEntity>(md => md.Index(indexName).Properties(GetIndexPropertiesDescriptor()));
                if (resp.IsValid)
                    return;
                throw new Exception($"更新索引映射失败，Error：{resp.GetErrorString()}", resp.OriginalException);
            }
        }

        /// <summary>
        /// 批量添加文档到ES
        /// </summary>
        /// <param name="documents">文档集合</param>
        /// <param name="refreshOnCompleted">强制在完成添加文档之后刷新索引，除非有足够的理由，否则请使用默认false</param>
        public void AddDocuments(IEnumerable<TEntity> documents, bool refreshOnCompleted = false)
        {
            if (documents == null || !documents.Any())
                return;
            var client = ElasticSearchHelper.GetElasticClient();
            var indexNames = documents.Select(x => x.GetIndexName()).Distinct().ToArray();
            var notExistsIndexNames = ElasticSearchHelper.ExtractNotExistsStatsIndices(indexNames);
            if (notExistsIndexNames != null && notExistsIndexNames.Length > 0)
            {
                foreach (var notExistsIndexName in notExistsIndexNames)
                {
                    CreateIndex(notExistsIndexName);
                }
            }
            var totalCount = documents.Count();
            var pageSize = 1000;
            var batchCount = (totalCount + pageSize - 1) / pageSize;
            for (int i = 0; i < batchCount; i++)
            {
                var batchDatas = documents.Skip(i * pageSize).Take(pageSize).ToArray();
                try
                {
                    var observableBulk = client.BulkAll(batchDatas, d => d.Index("")
                                            .MaxDegreeOfParallelism(4)
                                            .RefreshOnCompleted(refreshOnCompleted)
                                            .BackOffTime(TimeSpan.FromSeconds(10))
                                            .BackOffRetries(2)
                                            .Size(100)
                                            .BufferToBulk((descriptor, list) =>
                                            {
                                                foreach (var item in list)
                                                {
                                                    descriptor.Index<TEntity>(bi => bi
                                                        .Index(item.GetIndexName())
                                                        .Id(item.GetIndexDocumentId())
                                                        .Document(item)
                                                    );
                                                }
                                            }));
                    observableBulk.Wait(TimeSpan.FromSeconds(10), b => { });
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"添加文档到ES出错，documents:{JsonConvert.SerializeObject(batchDatas)}");
                    throw;
                }
            }
        }

        /// <summary>
        /// 检索一份文档
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="documentId">文档Id</param>
        /// <returns></returns>
        public async Task<TEntity> GetDocumentAsync(string indexName, string documentId)
        {
            var response = await ElasticSearchHelper.GetElasticClient().GetAsync<TEntity>(new GetRequest(indexName, documentId));
            if (response?.ApiCall == null || !response.ApiCall.Success)
                throw new Exception($"检索单个文档失败，IndexName：{indexName}，documentId：{documentId}，Error：{response.GetErrorString()}", response.OriginalException);
            if (!response.Found)
                return null;
            return response.Source;
        }

        /// <summary>
        /// 更新部分文档信息
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="_id">文档id</param>
        /// <param name="updateObj">构造的需要更新的字段对象</param>
		public void UpdateDocument<T>(string _id, object updateObj) where T : class, IDocumentIndexIdentity, new()
        {
            var model = new T();
            ElasticSearchHelper.GetElasticClient().Update<T, object>(_id, u => u.Index(model.GetIndexName()).Doc(updateObj));
        }

        /// <summary>
        /// 删除一个文档
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="documentId">文档Id</param>
        public async Task DeleteDocumentAsync(string indexName, string documentId)
        {
            var response = await ElasticSearchHelper.GetElasticClient().DeleteAsync(new DeleteRequest(indexName, documentId));
            if (response.IsValid || response.Result == Result.NotFound)
                return;
            throw new Exception($"删除单个文档失败，IndexName：{indexName}，documentId：{documentId}，Error：{response.GetErrorString()}", response.OriginalException);
        }

        /// <summary>
        /// 删除一个索引
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="documentId">文档Id</param>
        public bool DeleteDocument(string indexName)
        {
            var client = ElasticSearchHelper.GetElasticClient();
            var response = client.Indices.Delete(indexName);
                return response.IsValid;
            throw new Exception($"删除单个索引失败，IndexName：{indexName}，Error：{response.ServerError?.Error?.ToString()}", response.OriginalException);
        }

        /// <summary>
        ///复制索引数据到新索引
        /// </summary>
        /// <param name="indexName">Name of the index.</param>

        public bool CopyIndexData(string indexName, string copyIndex,int numberOfReplicas)
        {
            try
            {
                var client = ElasticSearchHelper.GetElasticClient();
                var reindex = client.Reindex<TEntity>(r => r
              .BackPressureFactor(10)
              .ScrollAll("1m", 2, s => s
                  .Search(ss => ss
                      .Index(indexName).TypedKeys())
                      .MaxDegreeOfParallelism(4))
              .CreateIndex(c => c
                  .Index(copyIndex)
                  .Settings(s => s.NumberOfShards(5).NumberOfReplicas(numberOfReplicas))
                  .Map<TEntity>(md => md.Properties(GetIndexPropertiesDescriptor())))
                  .BulkAll(b => b
                  .Index(copyIndex)
                  .Size(100)
                  .MaxDegreeOfParallelism(2)
                  .RefreshOnCompleted()));
                var o = new ReindexObserver(
               onError: (e) => { },
               onCompleted: () => { });
                reindex.Subscribe(o);
                return true;
            }
            catch (Exception e)
            {
                throw new Exception($"复制索引{indexName}数据到新索引{copyIndex}失败，IndexName：{indexName}，Error：{e.Message}");
            }

        }

        ///// <summary>
        ///// 大批量添加文档到ES（开发环境测试超过10000条会丢失部分数据）
        ///// </summary>
        ///// <param name="documents">文档集合</param>
        ///// <param name="refreshOnCompleted">强制在完成添加文档之后刷新索引，除非有足够的理由，否则请使用默认false</param>
        //public void AddDocuments(IEnumerable<TEntity> documents, bool refreshOnCompleted = false)
        //{
        //    if (documents == null || !documents.Any())
        //        return;
        //    var client = ElasticSearchHelper.GetElasticClient();
        //    var indexNames = documents.Select(x => x.GetIndexName()).Distinct().ToArray();
        //    var notExistsIndexNames = ElasticSearchHelper.ExtractNotExistsStatsIndices(indexNames);
        //    if (notExistsIndexNames != null && notExistsIndexNames.Length > 0)
        //    {
        //        foreach (var notExistsIndexName in notExistsIndexNames)
        //        {
        //            CreateIndex(notExistsIndexName);
        //        }
        //    }
        //    try
        //    {
        //        var observableBulk = client.BulkAll(documents, d => d.Index("")
        //                                .MaxDegreeOfParallelism(4)
        //                                .RefreshOnCompleted(refreshOnCompleted)
        //                                .BackOffTime(TimeSpan.FromSeconds(10))
        //                                .BackOffRetries(2)
        //                                .Size(100)
        //                                .BufferToBulk((descriptor, list) =>
        //                                {
        //                                    foreach (var item in list)
        //                                    {
        //                                        descriptor.Index<TEntity>(bi => bi
        //                                            .Index(item.GetIndexName())
        //                                            .Id(item.GetIndexDocumentId())
        //                                            .Document(item)
        //                                        );
        //                                    }
        //                                }));
        //        observableBulk.Wait(TimeSpan.FromSeconds(10), b => { });
        //    }
        //    catch (Exception ex)
        //    {
        //        Log.Write($"添加文档到ES出错，documents:{JsonConvert.SerializeObject(documents)}", MessageType.Error, GetType(), ex);
        //        throw;
        //    }
        //}

        /// <summary>
        /// 添加文档
        /// </summary>
        public void AddDocument(TEntity document)
        {
            AddDocuments(new List<TEntity> { document });
        }

    }
}
